package com.pengpeng.rocketmq.web;

import com.pengpeng.rocketmq.consumer.BatchConsumer;
import com.pengpeng.rocketmq.consumer.Consumer;
import com.pengpeng.rocketmq.producer.Producer;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.util.ArrayList;
import java.util.List;

/**
 * Describe:
 * Created by AlexP on 2022/4/25.
 */
@Slf4j
@RestController
@RequestMapping("/batch")
public class BatchController {
    @Autowired
    private Producer producer;

    @RequestMapping("/batchProducer")
    public Object batchProducer() throws Exception {
        List<Message> messages = new ArrayList<>();
        //总共发送五次消息
        for (int i = 0; i < 20; i++) {
            //创建生产信息
            Message message = new Message(BatchConsumer.TOPIC, "BatchTagA", ("Hello RocketMQ:" + i).getBytes(RemotingHelper.DEFAULT_CHARSET));
            messages.add(message);
        }
        //发送
        SendResult sendResult = producer.getProducer().send(messages);
        log.info("输出生产者信息={}",sendResult);
        return "成功";
    }

    /**
     * 仅当您发送大批量时，复杂性才会增加，您可能不确定它是否超过大小限制（1MiB）。此时，您最好拆分列表
     * @return
     * @throws Exception
     */
    @RequestMapping("/splitterProducer")
    public Object splitterProducer() throws Exception {
        List<Message> messages = new ArrayList<>();
        //总共发送五次消息
        for (int i = 0; i < 20; i++) {
            //创建生产信息
            Message message = new Message(BatchConsumer.TOPIC, "BatchTagA", ("Hello RocketMQ:" + i).getBytes(RemotingHelper.DEFAULT_CHARSET));
            messages.add(message);
        }
        //then you could split the large list into small ones:
        ListSplitter splitter = new ListSplitter(messages);
        while (splitter.hasNext()) {
            try {
                List<Message>  listItem = splitter.next();
                SendResult sendResult =producer.getProducer().send(listItem);
                log.info("输出生产者信息={}",sendResult);
            } catch (Exception e) {
                e.printStackTrace();
                //handle the error
            }
        }
        return "成功";
    }
}
